package com.ehualu.liaocheng

import java.text.SimpleDateFormat

import com.ehualu.liaocheng.until.OracleUtills
import com.typesafe.config.ConfigFactory
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}

import scala.collection.mutable.ArrayBuffer

/**
  * @author ：吴敬超
  */
object realTimekafka2 {


  def getDifferentSecondsu(start: String, end: String): Long = {

    val fm = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    val startLong = fm.parse(start).getTime
    val endLong = fm.parse(end).getTime

    val seconds = (endLong - startLong) / 1000
    seconds
  }

  val getDifferentSeconds = getDifferentSecondsu _

  def main(args: Array[String]): Unit = {

    //KAFKA安全认证
    security.security.run()

    //KAFKA安全认证
    //    security.run()
    //创建SparkConf，如果将任务提交到集群中
    val conf = new SparkConf().setAppName("realTimeMonitoringtest").setMaster("local[2]")
    //创建一个StreamingContext，批次时长为2秒
    val ssc = new StreamingContext(conf, Seconds(2))


    ssc.sparkContext.setLogLevel("WARN")
    //配置工厂 获取kakfa配置信息
    val load = ConfigFactory.load("kafka")

    //消费卡夫卡参数配置
    val kafkaParams = Map[String, Object](
      ///brokers 地址
      "bootstrap.servers" -> load.getString("kafka.brokers"),
      //指定该 consumer 将加入的消费组
      "group.id" -> load.getString("kafka.groupid"),
      //指定序列化类
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      // 是否开启自动提交 offset
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    //主题列表
    val topics = Array(load.getString("kafka.topics"))

    //直连方式消费
    val stream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](ssc,
      PreferConsistent,
      //订阅的策略（可以指定用正则的方式读取topic，比如my-ordsers-.*）
      Subscribe[String, String](topics, kafkaParams))


    val WindowsData: DStream[(String, (String, String))] = stream.map(

      line => {

        println("777777777777value777777777777：" + line.value())


        val arr: Array[String] = line.value().split(",")
        //      时间
        val sj = arr(3).substring(0, 19)

        //      号牌
        val hphm = arr(5)

        //      卡口编号
        val kkbh = arr(1)

        (hphm, (kkbh, sj))
      }

    ).filter(!_._1.toString.equals("车牌"))
//      .window(Seconds(60))
      .window(Minutes(1), Seconds(2))
//      .window(Seconds(60), Seconds(2))

//    val pvWindow = data.reduceByKeyAndWindow(_ + _, _ - _, Minutes(60), Minutes(10))

    //过滤窗口数据
    val filterHphmGroupedRDD = WindowsData.groupByKey().filter(e => {

      e._2.size > 1
    })
    //
    //    -------------------------------------------
    //    Time: 1567561488000 ms
    //      -------------------------------------------
    //    (鲁P1P118,ArrayBuffer((3715000121,2019-09-04 09:44:32), (3715000122,2019-09-04 09:44:26), (3715000736,2019-09-04 09:43:54)))
    //    (鲁P9166W,ArrayBuffer((37150002336,2019-09-04 09:44:20), (37150002335,2019-09-04 09:44:25)))
    //    (鲁P77R03,ArrayBuffer((3715000145,2019-09-04 09:44:25), (3715000147,2019-09-04 09:44:09)))
    //    (鲁PG1665,ArrayBuffer((3715000650,2019-09-04 09:44:44), (3715000651,2019-09-04 09:44:08)))
    //    (鲁P27P77,ArrayBuffer((3715000059,2019-09-04 09:44:20), (3715000060,2019-09-04 09:44:17)))
    //    (鲁AL3C79,ArrayBuffer((3715000176,2019-09-04 09:43:54), (3715001006,2019-09-04 09:43:20)))
    //    (鲁P302A5,ArrayBuffer((3715000754,2019-09-04 09:44:41), (3715000755,2019-09-04 09:44:38)))
    //    (鲁PSZ879,ArrayBuffer((3715000765,2019-09-04 09:44:08), (3715000764,2019-09-04 09:44:14)))
    //    (鲁P17M60,ArrayBuffer((3715000160,2019-09-04 09:43:57), (3715000159,2019-09-04 09:44:01)))
    //    (鲁P57913,ArrayBuffer((3715000290,2019-09-04 09:44:25), (3715000291,2019-09-04 09:44:20)))
    //    ...
    //
    //    -------------------------------------------
    //    Time: 1567561490000 ms
    //      -------------------------------------------
    //    (鲁P1P118,ArrayBuffer((3715000121,2019-09-04 09:44:32), (3715000122,2019-09-04 09:44:26), (3715000736,2019-09-04 09:43:54)))
    //    (鲁P9166W,ArrayBuffer((37150002336,2019-09-04 09:44:20), (37150002335,2019-09-04 09:44:25)))
    //    (鲁P77R03,ArrayBuffer((3715000145,2019-09-04 09:44:25), (3715000147,2019-09-04 09:44:09)))
    //    (鲁PG1665,ArrayBuffer((3715000650,2019-09-04 09:44:44), (3715000651,2019-09-04 09:44:08)))
    //    (鲁P27P77,ArrayBuffer((3715000059,2019-09-04 09:44:20), (3715000060,2019-09-04 09:44:17)))
    //    (鲁AL3C79,ArrayBuffer((3715000176,2019-09-04 09:43:54), (3715001006,2019-09-04 09:43:20)))
    //    (鲁P302A5,ArrayBuffer((3715000754,2019-09-04 09:44:41), (3715000755,2019-09-04 09:44:38)))
    //    (鲁PSZ879,ArrayBuffer((3715000765,2019-09-04 09:44:08), (3715000764,2019-09-04 09:44:14)))
    //    (鲁P17M60,ArrayBuffer((3715000160,2019-09-04 09:43:57), (3715000159,2019-09-04 09:44:01)))
    //    (鲁P57913,ArrayBuffer((3715000290,2019-09-04 09:44:25), (3715000291,2019-09-04 09:44:20)))
    //    ...

    val roadTimeRDD = filterHphmGroupedRDD.map(iter => {
      //            println("7777777777777777777777777777777777777777777")
      println("55555555车牌55555555:"+iter._1)


      var array = ArrayBuffer[(String, String)]()
      val iterator = iter._2.iterator
      //取出每个key 的对应的信息
      while (iterator.hasNext) {

        //        Iterator类的next( )方法在同一循环中不能出现两次。
        //        println(iterator.next())
        array.append(iterator.next())

      }

      val sortedTupleArray: ArrayBuffer[(String, String)] = array.sortBy(_._2)


      //      println(sortedTupleArray)
      val size: Int = sortedTupleArray.size

      println("444444size44444:"+size)
      var rearray = ArrayBuffer[(String, Long)]()
      for (i <- (0 to size - 2)) {

        var kkbh1 = sortedTupleArray(i)._1
        var kkbh2 = sortedTupleArray(i + 1)._1
        var kksj1 = sortedTupleArray(i)._2
        var kksj2 = sortedTupleArray(i + 1)._2


        println("****kkbh1******" + kkbh1)
        println("****kkbh2******" + kkbh2)
        println("****kksj1******" + kksj1)
        println("****kksj2******" + kksj2)
        var ldbh = kkbh1 + "_" + kkbh2
        var ldsj = getDifferentSeconds(kksj1, kksj2)


        println("****ldbh******" + ldbh)
        println("****ldsj******" + ldsj)
        var ldyz = (ldbh, ldsj)
        rearray.append(ldyz)
      }
      rearray


    }

    )

    //    roadTimeRDD.print()


    val notNullRoadRDD = roadTimeRDD.filter(e => {
      e.size > 0
    })

    println("8888888888888888888888888888888888888")

    //    notNullRoadRDD.foreach(println)
    //
    //    notNullRoadRDD.count()

    //
    //    (37150002324_37150002323,2)
    //    (37150002324_37150002323,22)
    //    (37150002324_37150002323,4)
    //    (37150002324_37150002323,3)
    //    (37150002324_37150002323,9)
    //    (3715000916_3715000916,15491)
    //    (3715000916_3715000916,2943)
    //    (3715000916_3715000916,2897)
    //    (3715000916_3715000916,11628)
    //    (3715000916_3715000916,6858)
    //    (3715000916_37150002324,41518)
    //


    //    notNullRoadRDD.flatMap(x=>x).foreach(println(_))

    //
    //    (37150002322_37150002322,CompactBuffer(167, 25229, 33481, 32468, 25178, 36636, 13627, 20488, 19096, 7057, 4669, 19367, 20098, 35439, 14731, 24269, 29231, 5950, 25766, 14665, 24191, 22960, 34038, 30073, 17, 41892, 16020, 6184, 16396, 28850, 9172, 2292, 24253, 5772, 18928, 59418, 22210, 24608, 45960, 14303, 22250, 19074, 12824, 16838, 19949, 15463, 4391, 17861, 11023, 21390, 26700, 5527, 46507, 43468, 27961, 22408, 12614, 11622, 23470, 38952, 19334, 22571, 41598, 13169, 31160, 4173, 7389, 67684, 35974, 23482, 22431, 10499, 35127, 17116, 26324, 17978))
    //    (3715000916_37150002323,CompactBuffer(30728, 9082, 33018, 24168, 4956, 6550, 37948, 6846, 6018))
    //    (37150002323_37150002323,CompactBuffer(2081, 13736, 22435, 18363, 38447, 21672, 9596, 32641, 32178, 28, 3869, 77995, 39852, 13141, 8219, 10270, 24335, 10199, 6829, 13247, 14133, 3588, 8333, 41620, 26961, 28715, 4251))
    //notNullRoadRDD.flatMap(x => x).groupByKey()
    val luDuanRdd = notNullRoadRDD.flatMap(x => x).groupByKey()

    val luduantjrdd = luDuanRdd.map(

      line => {
        println("&&&&&&&&路段编号&&&&&&&&&" + line._1)

        println("&&&&&&&&line._2&&&&&&&&&" + line._2.toString())

        //        line._1
        var sjnum = line._2.iterator.size

        println("*******sjnum******" + sjnum)

        val sjiterator = line._2.iterator
        var sumsj: Long = 0
        //取出每个key 的对应的信息
        while (sjiterator.hasNext) {
          sumsj = sumsj + sjiterator.next()
          //        Iterator类的next( )方法在同一循环中不能出现两次。
          //        println(iterator.next())

        }
        var ldpjsj = sumsj / sjnum

        println("********路段平均时间***************：" + ldpjsj)

        (line._1, ldpjsj)
      }
    )


    println("99999999999999999999999999999999999999999")
    luduantjrdd.print()


    ssc.start()
    //等待退出
    ssc.awaitTermination()

  }
}
